// Copyright 2020 newtbig Author. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package hash

import (
	"errors"
	"hash/crc32"
	"sync"
)

const (
	REPLICASNUMVir = 5
	RING_LENTH     = 1024
	//RING_ID_MAX    = uint32(^uint32(0) >> 1) //4294967295
)

var ErrEmptyRingVir = errors.New("empty ring")

type (
	HashRingVir struct {
		m_Length    int
		m_Nodes     []uint32
		m_RingMap   map[uint32]*Node
		m_MemberMap map[string]int
		sync.RWMutex
	}
	Node struct {
		Key string
		Set bool
	}
)

func NewHashRingVir(n int) *HashRingVir {
	if n <= 0 {
		return nil
	}
	pRing := new(HashRingVir)
	pRing.m_Length = n
	pRing.m_MemberMap = make(map[string]int)
	pRing.initNodes()
	return pRing
}

func (this *HashRingVir) add(n int, elt string) bool {
	if n < 0 || n >= this.m_Length {
		return false
	}
	set := false
	for i := n; i >= 0; i-- {
		id := this.m_Nodes[i]
		if this.m_RingMap[id].Set {
			break
		}
		this.m_RingMap[id].Key = elt
		if !set {
			this.m_RingMap[id].Set = true
			set = true
		}
	}
	if set {
		this.m_MemberMap[elt] = n
	}
	return set

}

func (this *HashRingVir) reset() {
	for _, node := range this.m_RingMap {
		node.Key = ""
		node.Set = false
	}
}

func (this *HashRingVir) remove(elt string) bool {
	n, ok := this.m_MemberMap[elt]
	if !ok {
		return false
	}
	delete(this.m_MemberMap, elt)
	if len(this.m_MemberMap) == 1 {
		this.reset()
		return true
	}

	this.m_RingMap[this.m_Nodes[n]].Set = false
	next := n + 1
	if next == this.m_Length {
		next = 0
	}
	key := this.m_RingMap[this.m_Nodes[next]].Key

	for i := n; i >= 0; i-- {
		id := this.m_Nodes[i]
		if this.m_RingMap[id].Set {
			break
		}
		this.m_RingMap[id].Key = key
	}

	return true

}

func (this *HashRingVir) hashKey(key string) uint32 {
	if len(key) < 64 {
		var scratch [64]byte
		copy(scratch[:], key)
		return crc32.ChecksumIEEE(scratch[:len(key)])
	}
	return crc32.ChecksumIEEE([]byte(key))
}

func (this *HashRingVir) Add(n int, elt string) {
	this.Lock()
	defer this.Unlock()
	this.add(n, elt)
}

func (this *HashRingVir) Remove(elt string) {
	this.Lock()
	defer this.Unlock()
	this.remove(elt)
}

func (this *HashRingVir) HasMember(elt string) bool {
	this.RLock()
	defer this.RUnlock()
	_, bEx := this.m_MemberMap[elt]
	return bEx
}

func (this *HashRingVir) Members() []string {
	this.RLock()
	defer this.RUnlock()
	var m []string
	for k := range this.m_MemberMap {
		m = append(m, k)
	}
	return m
}

func (this *HashRingVir) Get(name string) (error, string) {
	this.RLock()
	defer this.RUnlock()
	if len(this.m_MemberMap) == 0 {
		return ErrEmptyRingVir, ""
	}
	key := this.hashKey(name)
	i := key / uint32(this.m_Length)
	if i > 0 {
		i = i - 1
	}
	for ; i < uint32(this.m_Length); i++ {
		if this.m_Nodes[i] >= key {
			return nil, this.m_RingMap[this.m_Nodes[i]].Key
		}
	}

	return ErrEmptyRing, ""
}

//
func (this *HashRingVir) initNodes() {
	n := this.m_Length
	this.m_Nodes = make([]uint32, n)
	this.m_RingMap = make(map[uint32]*Node, n)

	maxUint32 := uint32(^uint32(0) >> 1)
	baseInterval := maxUint32 / uint32(n)
	extraUnits := maxUint32 % uint32(n)

	current := uint32(0)
	for i := 0; i < n; i++ {
		this.m_Nodes[i] = current
		this.m_RingMap[current] = &Node{Key: "", Set: false}
		if extraUnits > 0 {
			current += baseInterval + 1
			extraUnits--
		} else {
			current += baseInterval
		}
	}
	return
}
